package cn.middle.remote.user;


import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import cn.middle.remote.Buffer;
import cn.middle.remote.exception.ResponseException;
import cn.middle.remote.exchange.DefaultFuture;
import cn.middle.remote.message.BaseSignal;
import cn.middle.remote.message.EdgeReplySignal;
import cn.middle.remote.message.EventReportSignal;
import cn.middle.remote.message.HeartbeatPongSignal;
import cn.middle.remote.message.LoginResultSignal;
import cn.middle.remote.message.ReplyMessageSignal;
import cn.middle.remote.message.SignalHelper;
import cn.middle.remote.util.NettyAttributeUtil;
import cn.middle.remote.util.Util;
import io.netty.channel.ChannelHandlerContext;

public class NettyUserProcessHandler extends UserProcessHandler {
	
	private static final Logger log = LoggerFactory.getLogger(NettyUserProcessHandler.class);

	
	ExecutorService  executors = Executors.newFixedThreadPool(10);

	@Override
	public void channelInactive(ChannelHandlerContext ctx) throws Exception {
		this.listener.connectLost(NettyAttributeUtil.clientId(ctx.channel()),1);
		log.error("user-close ->" + ctx.channel().remoteAddress());

	}

	@Override
	public void received(ChannelHandlerContext ctx, BaseSignal msg) throws Exception {
		executors.execute(() -> {
			byte msgType = msg.getMessageType();
			switch(msgType) {
			case SignalHelper.REPLY:
				reply((ReplyMessageSignal) msg);
				break;
			case SignalHelper.ACK:
				ack(msg.getId());
				break;
			case SignalHelper.PONG:
				pong((HeartbeatPongSignal)msg);
				break;
			case SignalHelper.EVENT_REPORT:
				event((EventReportSignal)msg);
				break;
			case SignalHelper.EDGE_REPLY:
				edgeReply((EdgeReplySignal)msg);
				break;
			case SignalHelper.LOGIN_RESULT:
				login(ctx,(LoginResultSignal) msg);
				break;
			}
		});
	}
	
	private void ack(String msgId) {
		//DefaultFuture.ack(msgId);
		log.info("ack -> {}" ,msgId );
		this.listener.ack(msgId);
	}
	
	private void edgeReply(EdgeReplySignal signal) {
		listener.edgeReply(new Buffer(signal.getId(), signal.getData(), serialization));
	}
	
	private void event(EventReportSignal signal) {
		listener.subscribeEvent(signal.getEventId(), new Buffer(signal.getId(),signal.getData(),serialization));
	}
	
	private void login(ChannelHandlerContext ctx,LoginResultSignal signal) {
		listener.login(signal.getCode(), signal.getMessage());
	}
	
	private void reply(ReplyMessageSignal msg) {
		if(msg.getCode() != 0) {
			DefaultFuture.setCause(msg.getId(), new ResponseException(Util.isNotNull(msg.getMessage()) ? msg.getMessage() : "未知异常"));
		}else {
			DefaultFuture.callResponse(msg.getId(), msg.getData());
		}
		listener.reply(msg.getId(), msg.getInterfaceName(), new Buffer(msg.getId(), msg.getData(), serialization));
	}
	
	public void pong(HeartbeatPongSignal pong) {
		log.info(pong.getId() + " : " + HeartbeatPongSignal.PONG);
	}

	@Override
	public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
		log.error(cause.getMessage(),cause);
		listener.exception(cause);
	}

}
